package com.boot.study.udf

import com.boot.study.api.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.functions.ScalarFunction
import org.apache.flink.types.Row

object ScalarFunctionTest {
  def main(args: Array[String]): Unit = {
    // 1: 创建环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) // 时间语义，处理时间

    // 创建表执行环境
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)

    val inputPath: String = "D:\\WorkSpace\\idea\\Flink\\src\\main\\resources\\sensor.txt"
    val inputSteam: DataStream[String] = env.readTextFile(inputPath)
    val dataStream: DataStream[SensorReading] = inputSteam.map(data => {
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
      // 延迟1秒生成 watermark
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(element: SensorReading): Long = element.timeStamp * 1000
      })
    val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'temperature, 'timeStamp.rowtime as 'ts)

    // 调用自定义hash函数，对id进行hash运算
    // 1.1 table
    // 首先new udf实例
    val hashCode = new HashCode(20)
    val resultTable: Table = sensorTable
      .select('id, 'ts, hashCode('id))

    // 1.2 sql
    // 需要在表环境中注册
    tableEnv.createTemporaryView("sensor", sensorTable)
    tableEnv.registerFunction("hashcode", hashCode)
    val resultSqlTable: Table = tableEnv.sqlQuery(
      """
        |select id, ts, hashcode(id) from sensor
        |""".stripMargin)

    // 转换流输出
    resultTable.toAppendStream[Row].print("table")
    resultSqlTable.toAppendStream[Row].print("sql")

    env.execute("scalar function test")
  }
}

/**
 * 自定义标量函数
 */
class HashCode(factor: Int) extends ScalarFunction {
  def eval(s: String): Int = {
    s.hashCode * factor - 10000
  }
}